package com.cn.broker;

import com.alibaba.fastjson.JSONObject;
import com.cn.broker.processor.BrokerAdminProcessor;
import com.cn.broker.processor.ClientProcessor;
import com.cn.broker.processor.MessageCommitProcessor;
import com.cn.broker.processor.PullRequestProcessor;
import com.cn.common.BrokerEntity;
import com.cn.common.ClientType;
import com.cn.common.DataSizeUtil;
import com.cn.common.GlobalConfigUtil;
import com.cn.common.consumer.ConsumerOffsetManager;
import com.cn.common.processor.*;
import com.cn.common.topic.TopicManager;
import com.cn.example.rpc.IDemoService;
import com.cn.example.rpc.impl.DemoService;
import com.cn.exception.MyException;
import com.cn.file.DefaultMessageStore;
import com.cn.file.MessageStoreConfig;
import com.cn.msg.Message;
import com.cn.remote.ChannelListener;
import com.cn.remote.RemoteClient2;
import com.cn.remote.RemoteServer;
import com.cn.rpc.RpcServer;
import com.cn.util.*;
import org.apache.commons.lang3.StringUtils;

import java.io.File;
import java.nio.charset.Charset;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;

public class BrokerController {

    private ClientType clientType;

    private RemoteClient2 client;


    private RemoteServer brokerServer;

    private FileConfigPropertyUtil config;

    private int port;
    private String host;
    private String name;
    private String address;
    private String cluster_name;
    private int broker_id;

    //可以挂起15秒
    private long pullSuspendTimeStamp = 15000;
    /**
     * 更新topic
     */
    private ScheduledExecutorService pullTopicExecutorService;

    /**
     * 更新集群
     */
    private ScheduledExecutorService pullClusterExecutorService;

    /**
     * 定时检测是否连接到主节点
     */
    private ScheduledExecutorService restartConnect2Master;

    /**
     *
     */
    private ScheduledExecutorService registerTopicExecutorService;


    private ScheduledExecutorService registerLoginExecutorService;

    private ScheduledExecutorService defaultScheduledExecutorService;

    private ExecutorService defaultExecutorService = Executors.newFixedThreadPool(16);

    /**
     * nameSrv 服务节点地址
     */
    private String nameSrv;

    private final BrokerEntity brokerEntity;

    private DefaultMessageStore defaultMessageStore;

    private ConsumerOffsetManager consumerOffsetManager;

    private RebalanceLockManager rebalanceLockManager;

    private PullRequestProcessor pullProcessor;
    private PullMessageRequestHoldService pullMessageRequestHoldService;

    private BlockingQueue<Runnable> pullBlockQueue = new LinkedBlockingQueue<>(10000);
    private ExecutorService pullQueueExecutorService;

    private BlockingQueue<Runnable> commitLogQueue = new LinkedBlockingQueue<>(100000);
    private ExecutorService commitLogExecutorService;

    private BlockingQueue<Runnable> clientQueue = new LinkedBlockingQueue<>(10000);
    private ExecutorService clientExecutorService;

    private TopicManager topicManager;
    private ConsumerManager consumerManager;
    private ChannelListener channelListener;


    public BrokerController(FileConfigPropertyUtil config) {
        if (config == null) {
            throw new MyException("配置文件错误！");
        }
        this.config = config;

        this.nameSrv = this.config.get("namesrv_address");
        this.port = Integer.valueOf(this.config.get("broker_port"));
        this.host = this.config.get("broker_address");
        this.name = this.config.get("broker_name");
        this.address = String.format("%s:%d", host, port);
        String broker_id = this.config.get("broker_id");
        this.cluster_name = this.config.get("cluster_name");


        brokerEntity = new BrokerEntity();
        brokerEntity.setName(this.name).setAddress(this.address).setCluster_name(cluster_name);
        if (StringUtils.isNotEmpty(broker_id)) {
            brokerEntity.setBroker_id(Integer.valueOf(broker_id));
            this.broker_id = brokerEntity.getBroker_id();
        }

        //数据目录
        String data_dir = this.config.get("data_dir");
        if (StringUtils.isEmpty(data_dir)) throw new MyException("请配置数据目录");

        //根目录
        String root_dir = this.config.get("root_dir");

        MessageStoreConfig messageStoreConfig = new MessageStoreConfig(root_dir, data_dir + File.separator, DataSizeUtil.getDataBaseSize("1g"));
        messageStoreConfig.setFlushDiskType(this.config.get("flush_way"));

        this.defaultMessageStore = new DefaultMessageStore(messageStoreConfig, this);

        //consumeroffset
        String consumerOffset = root_dir + "/config/consumerOffset.json";
        this.consumerOffsetManager = new ConsumerOffsetManager(consumerOffset);

        //注册 namesrv
        this.client = new RemoteClient2(null);

        //实例化 broker节点
        this.brokerServer = new RemoteServer(32, this.address);
        this.channelListener = new ClientHouseChannelListener(this);
        this.brokerServer.registerChannelListener(channelListener);

        //初始化 topic拉取服务
        this.pullTopicExecutorService = Executors.newScheduledThreadPool(1);

        //初始化集群服务(拉取集群的 url信息)
        this.pullClusterExecutorService = Executors.newScheduledThreadPool(1);

        //更新topic
        this.registerTopicExecutorService = Executors.newScheduledThreadPool(1);

        this.registerLoginExecutorService = Executors.newScheduledThreadPool(1);

        this.defaultScheduledExecutorService = Executors.newScheduledThreadPool(1);
        this.pullProcessor = new PullRequestProcessor(this);
        this.pullMessageRequestHoldService = new PullMessageRequestHoldService(this);
        this.pullQueueExecutorService = new ThreadPoolExecutor(GlobalConfigUtil.queueCoreMin, GlobalConfigUtil.queueCoreMax, GlobalConfigUtil.coreActiveLive, TimeUnit.MILLISECONDS, this.pullBlockQueue);

        this.commitLogExecutorService = new ThreadPoolExecutor(GlobalConfigUtil.queueCoreMin, GlobalConfigUtil.queueCoreMax, GlobalConfigUtil.coreActiveLive, TimeUnit.MILLISECONDS, this.commitLogQueue);

        String topicPath = root_dir + "/config/topic.json";
        this.topicManager = new TopicManager(topicPath);

        this.rebalanceLockManager = new RebalanceLockManager();
        this.consumerManager = new ConsumerManager();
        this.clientExecutorService = new ThreadPoolExecutor(5, 500, GlobalConfigUtil.coreActiveLive, TimeUnit.MILLISECONDS, clientQueue);

    }

    public void load() {
        this.defaultMessageStore.load();
        this.defaultMessageStore.start();
        //扫描接口
        scanRpc();
    }

    /**
     * 启动服务
     *
     * @throws InterruptedException
     */
    public void start() throws InterruptedException {

        this.load();

        this.client.start();
        this.brokerServer.start();
        this.pullMessageRequestHoldService.start();

        //注册broker基本信息
        this.registerLoginExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                //说明是broker节点，向namesrv节点注册broker信息
                Message message = new Message();
                message.setType((byte) 2);//broker节点注册
                message.setFlag((byte) 1);//表示向namesrv发送的请求
                message.setByteBodys(GsonUtil.converObj2Json(getBrokerEntity()).getBytes(Charset.forName("utf-8")));
                client.invokeMsg2(message, nameSrv);

            }
        }, 1, 3, TimeUnit.SECONDS);


        this.pullTopicExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                Message message = new Message();
                message.setType((byte) 4);//拉取topic
                message.setFlag((byte) 1);
                message.setByteBodys("topic".getBytes(Charset.forName("utf-8")));

                client.invokeMsg2(message, nameSrv);
            }
        }, 3, 5, TimeUnit.SECONDS);

        this.pullClusterExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                pullMasterSlave();
            }
        }, 5, 30, TimeUnit.SECONDS);

        if (this.broker_id > 0) {
            this.restartConnect2Master = Executors.newScheduledThreadPool(1);
            this.restartConnect2Master.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    connect2Master();
                }
            }, 3, 5l, TimeUnit.SECONDS);
        }

        //持久化位移
        this.defaultScheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                consumerOffsetManager.persit();
            }
        }, 1, 5, TimeUnit.SECONDS);

        //推送topic信息
        this.defaultScheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                Message message = new Message();
                message.setFlag((byte) 1);
                message.setType((byte) 14);
                message.setNettyRequestWrapperCode((byte) ProcessorCode.pull_default);
                message.setByteBodys(topicManager.getJson().getBytes(Charset.forName("utf-8")));
                message.setLength(message.getByteBodys().length);
//                System.out.println(nameSrv);
                client.invokeMsg2(message, nameSrv);
            }
        }, 10, 60 * 1, TimeUnit.SECONDS);//1分钟

        //注册服务
        registerProcessor();
    }

    /**
     * 关闭服务
     */
    public void stop() {
        if (this.client != null) {
            this.client.shutdown();
        }
        if (this.brokerServer != null) {
            this.brokerServer.shutdown();
        }
        this.pullTopicExecutorService.shutdown();
        this.defaultMessageStore.stop();
        this.pullMessageRequestHoldService.stop();
    }

    public void scanRpc() {
        IDemoService demoService = new DemoService();
        RpcServer.put(IDemoService.class.getName(), demoService);
    }

    /**
     * slave连接到Master
     */
    public void connect2Master() {

    }

    public void pullMasterSlave() {

        if (true) {
            Message message = new Message();
            message.setType((byte) 6);//拉取集群信息
            message.setFlag((byte) 1);
            message.setByteBodys(cluster_name.getBytes(Charset.forName("utf-8")));
            byte[] result = client.invokeMsg(message, nameSrv, 1000);//1秒超时
            if (result != null) {
                Set<BrokerEntity> set = new HashSet<BrokerEntity>();
                Set<JSONObject> s1 = RemotingSerializable.decode(result, HashSet.class);
                try {
                    for (JSONObject s : s1) {
                        set.add(s.toJavaObject(BrokerEntity.class));
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                if (set != null) {
                    this.defaultMessageStore.addClusterTables(cluster_name, set);
                }
            }
        }

    }

    public void registerProcessor() {
        //注册消息拉取服务

        this.brokerServer.registerProcessor(ProcessorCode.pull_msg, pullProcessor, this.defaultExecutorService);

        //消息位移拉取服务
        INettyRequestProcessor defaultProcessor = new BrokerAdminProcessor(this);
        this.brokerServer.registerProcessor(ProcessorCode.pull_offset, defaultProcessor, pullQueueExecutorService);

        //存储消息服务
        INettyRequestProcessor messageStoreProcessor = new MessageCommitProcessor(this);
        this.brokerServer.registerProcessor(ProcessorCode.push_msg, messageStoreProcessor, this.commitLogExecutorService);

        //
        INettyRequestProcessor clientProcessor = new ClientProcessor(this);
        this.brokerServer.registerProcessor(ProcessorCode.hearbeat, clientProcessor, this.clientExecutorService);

    }

    public PullMessageRequestHoldService getPullMessageRequestHoldService() {
        return pullMessageRequestHoldService;
    }

    public ExecutorService getDefaultExecutorService() {
        return defaultExecutorService;
    }

    public PullRequestProcessor getPullProcessor() {
        return pullProcessor;
    }

    public DefaultMessageStore getDefaultMessageStore() {
        return defaultMessageStore;
    }

    public ConsumerOffsetManager getConsumerOffsetManager() {
        return consumerOffsetManager;
    }

    public BrokerEntity getBrokerEntity() {
        return brokerEntity;
    }

    public long getPullSuspendTimeStamp() {
        return pullSuspendTimeStamp;
    }

    public ConsumerManager getConsumerManager() {
        return consumerManager;
    }

    public void setPullSuspendTimeStamp(long pullSuspendTimeStamp) {
        this.pullSuspendTimeStamp = pullSuspendTimeStamp;
    }

    public RebalanceLockManager getRebalanceLockManager() {
        return rebalanceLockManager;
    }

    public TopicManager getTopicManager() {
        return topicManager;
    }

    public int getPort() {
        return port;
    }

    public String getHost() {
        return host;
    }

    public String getName() {
        return name;
    }

    public String getAddress() {
        return address;
    }

    public String getCluster_name() {
        return cluster_name;
    }

    public int getBroker_id() {
        return broker_id;
    }
}
